package com.xxd.dfs.namenode;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-08-14 09:16
 * @description: editlog 操作组件
 */
public class FSEditLog {

    private final static Logger logger = LoggerFactory.getLogger(FSEditLog.class);

    private EditLogDoubleBuffer doubleBuffer = new EditLogDoubleBuffer();

    private FSNameSystem nameSystem;

    long curTxid = 0L;

    public volatile boolean isSyncScheduledToDisk = false;

    public volatile boolean isSyncRunning = false;

    private volatile long synctxid = 0L;

    ThreadLocal<Long> threadLocal = new ThreadLocal<>();

    private Timer editLogCleanerTimer = new Timer("FSEditLog-CleanerTimer", true);

    Configuration conf = Configuration.getInstance();

    public FSEditLog(FSNameSystem nameSystem) {
        this.nameSystem = nameSystem;
        // this.editLogCleanerTimer.schedule(new EditLogCleaner(), 3 * 60 * 1000, 3 * 60 * 1000);
        this.editLogCleanerTimer.schedule(new EditLogCleaner(),
                60 * 1000,
                60 * 1000);
    }

    /**
     * 1、先往双层缓存里面写入
     * 2、在往磁盘文件同步
     *
     * @param path
     */
    public void logEdit(String path, String editLogOp) throws IOException {
        // 假设 50条记录刷一次，那么就会说 假设同时有60条请求过来了前49条没有问题
        // 第50条过来了发现需要调度刷盘了，然后会把 isSyncScheduledToDisk = true;
        // 第51-60条 就会hang在那里
        // 在来50 61-110 条的话，假设说此时刚好调度刷盘完成，那么第61-99 就很快执行完毕
        // 第100条的时候
        synchronized (this) {
            logger.info("当前的 cur Txid = {}", curTxid);
            // 这个功能是想给磁盘调度线程让让路，我没调度好，你别往我的缓冲区里面写了，我吃不消呀
            // 这里卡了 101-200 的Txid
            waitIfAutoSyncScheduled();

            EditLog editLog = new EditLog(curTxid++, editLogOp, path);
            threadLocal.set(curTxid);

            doubleBuffer.write(editLog);

            // 控制一下需要调用的执行逻辑
            if (!doubleBuffer.shouldSyncToDisk()) {
                return;
            }
            logger.info("=============内存双缓冲满了需要调度刷盘=================");
            isSyncScheduledToDisk = true;
        }

        logSync();
    }


    /**
     * 同步数据到磁盘
     * 1、交换双层缓冲
     * 2、刷入磁盘
     */
    private void logSync() throws IOException {
        // 千辛万苦抢到时间片了
        synchronized (this) {
            long threadTxid = threadLocal.get();

            while (threadTxid > synctxid && isSyncRunning) {
                // 假设这里卡了 100 的那个线程
                // 当notify的时候，我不争气一直没有抢到时间片，导致可能150的线程都进来了
                // ok，那是不是需要做一下并发控制，比如说50抢到，那100在抢到是没有问题
                // 就怕 我100都抢到你50就别抢了 ==> 引出一个 线程对应的id概念
                try {
                    wait(2000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

            if (threadTxid < synctxid) {
                logger.info("当前线程的txid = {} 比 synctxid = {} 小，没有必要刷盘了", threadTxid, synctxid);
                return;
            }

            doubleBuffer.setReadyToFlush();
            synctxid = threadTxid;

            isSyncScheduledToDisk = false;
            isSyncRunning = true;
            notifyAll();
        }

        // 假设这个刷盘过程很耗时
        // 那么会导致说，我这次都还没刷完呢，第二次刷盘的线程又来了
        // 这样肯定是不行的必须控制一下并发
        // 假设TXid = 50 卡在这里
        // 这个时候 Txid = 100 就来刷了发现 怎么 50 还在刷呢
        // 可能 150 也都要过来了 是吧
        // 那么我肯定要保证同一时间只有一个线程刷磁盘，此时引入一个变量 isSyncRunning 来保证一下，判断一下咯
        doubleBuffer.flushToDisk();

        synchronized (this) {
            isSyncRunning = false;
            notifyAll();
        }
    }

    /**
     * 强制将edit log 刷入磁盘
     *
     * @throws IOException
     */
    public void forceFlush() throws IOException {
        synchronized (this) {
            doubleBuffer.setReadyToFlush();
            doubleBuffer.flushToDisk();
        }
    }

    /**
     * 等待调度刷盘逻辑
     */
    private void waitIfAutoSyncScheduled() {
        while (isSyncScheduledToDisk) {
            try {
                wait(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void setEditLogFiles(List<String> fileNames) {
        doubleBuffer.setEditLogFiles(fileNames);
    }

    public List<String> getEditLogFiles() {
        return doubleBuffer.getEditLogFiles();
    }

    public void removeEditLogFile(String editLogFile) {
        doubleBuffer.removeEditLogFile(editLogFile);
    }

    public long getCheckpointTxid() {
        return nameSystem.getCheckpointTxid();
    }

    public void setSyTxid(long checkpointTxid) {
        this.curTxid = checkpointTxid + 1;
        doubleBuffer.setStartTxid(checkpointTxid);
    }

    /**
     * 清理startTxid 比 checkpoint id 少的文件
     */
    class EditLogCleaner extends TimerTask {
        public EditLogCleaner() {
            logger.info("定时清理edit log 文件线程启动");
        }

        @Override
        public void run() {
            refreshEditLogFile();
        }
    }

    void refreshEditLogFile() {
        long checkpointTxid = getCheckpointTxid();
        List<String> editLogFiles = getEditLogFiles();
        logger.info("editLogFiles = {} checkpointTxid = {}", editLogFiles, checkpointTxid);
        for (String editLogFile : editLogFiles) {
            long startTxid = Long.parseLong(editLogFile.split("-")[0].split("_")[1]);
            long endTxid = Long.parseLong(editLogFile.split("-")[1].split("\\.")[0]);
            if (checkpointTxid >= endTxid) {
                deleteEditFile(editLogFiles, editLogFile);
            }
        }
    }

    private void deleteEditFile(List<String> editLogFiles, String editLogFile) {
        logger.info("edit log 文件 = {} 没有作用了 正在删除", editLogFile);
        removeEditLogFile(editLogFile);
        editLogFiles.remove(editLogFile);
        String dir = conf.get("editlog.dir");
        try {
            Files.delete(Paths.get(dir + editLogFile));
        } catch (IOException e) {
            logger.error("error", e);
        }
    }
}